package com.lz.dsp.task.config.kafka;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;


@Data
@Component
@ConfigurationProperties(prefix = "kafka.consumer.listener")
@DependsOn("init")
@Slf4j
public class KafkaConsumerListenerInfo {
    private String concurrencyInfo = "ad-dsp_creative-sync-group:1:false,ad-dsp_put_unit_dev-sync-group:1:false,ad-ssp_ad_slot_rtb-sync-group:1:false";
    private Map<String, ListenerDetails> listenerConfMap = new HashMap<>();

    @PostConstruct
    private void init() {
        if (StringUtils.isNotBlank(concurrencyInfo)) {
            String[] elementArray = concurrencyInfo.split(",");
            if (elementArray != null && elementArray.length > 0) {
                for (String element : elementArray) {
                    String[] elementInfo = element.split(":");
                    if (elementInfo != null && elementInfo.length == 3) {
                        listenerConfMap.put(elementInfo[0], new ListenerDetails(Integer.parseInt(elementInfo[1]), StringUtils.equals("true", elementInfo[2])));
                    }
                }
            }
        }
        if (listenerConfMap.isEmpty()) {
            log.info("kafka consumer listener container factory config is null.");
            return;
        }
        // 打印信息
        StringBuilder sb = new StringBuilder();
        sb.append("\nkafka consumer listener container factory config: \n");
        for (Map.Entry<String, ListenerDetails> entry : listenerConfMap.entrySet()) {
            sb.append("consumer group id : " + entry.getKey() + " | concurrency : " + entry.getValue().getConcurrencyNum() + " | isBatchFetch : " + entry.getValue().isBatchFetch() + " \n");
        }
        log.info(sb.toString());
    }
}

@Data
class ListenerDetails {
    private Integer concurrencyNum;
    private boolean isBatchFetch;

    public ListenerDetails(Integer concurrencyNum, boolean isBatchFetch) {
        this.concurrencyNum = concurrencyNum;
        this.isBatchFetch = isBatchFetch;
    }
}
